package com.wj.dubbo.service.task;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.wj.console.rabbit.message.CrawlerSocialFundMessage;
import com.wj.dubbo.utils.ThreadUtils;

public abstract class AbstractTaskService implements ITaskService {
	private static final Logger log = LoggerFactory.getLogger(AbstractTaskService.class);
	private volatile AtomicBoolean pause = new AtomicBoolean(false);
	
	private static int DEFAULT_NTHREADS = 2;//任务处理并发线程数
	
	private ExecutorService executor = null;
	
	public AbstractTaskService(int nThreads){
		executor = ThreadUtils.newFixedPoolSizeThreadPool(nThreads <= 0 ? DEFAULT_NTHREADS : nThreads);
	}
	
	public AbstractTaskService(){
		executor = ThreadUtils.newFixedPoolSizeThreadPool(DEFAULT_NTHREADS);
	}
	
	@Override
	public void pauseTask(CrawlerSocialFundMessage message) {
		log.info("暂停{}", serviceName());
		pause.compareAndSet(false, true);
		//如果有信息，则重新放回消息队列
		if(null != message){
			processPauseMessage(message);
		}
		
	}

	/**
	 * 服务暂停时消息处理（可以放入消息队列或者丢弃）
	 * @param message
	 */
	protected void processPauseMessage(CrawlerSocialFundMessage message){
		
	}

	@Override
	public void startTask() {
		log.info("启动{}", serviceName());
		pause.compareAndSet(true, false);
	}
	
	@Override
	public void pause(int minute){
		if(isPaused()){
    		try {
    			Thread.sleep(1000L*60*minute);
    		} catch (InterruptedException e) {
    			
    		}
    	}
	}
	
	public boolean isPaused(){
		return pause.get();
	}
	
	@Override
	public <T extends CrawlerTask> void submitTask(T task){
		//提交任务，如果处理任务已经达到最大时，则阻塞
		executor.execute(task);
	}
	/**
	 * 服务名称
	 * @param name
	 */
	protected abstract String serviceName();

}
